package test;

import java.util.HashSet;
import java.util.Scanner;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import preprocess.DumpSaler;

import trouger.dmt.DumpTableRunner;
import trouger.dmt.SourceRecordFilter;
import utils.Utils;

public class ExtractUserByName implements Tool{

	public static String getExtractUserByNamePath(String date) {
		return Utils.rootDir + date + "/ExtractUserByName/";
	}
	
	public static class SourceFilter implements SourceRecordFilter {
		private static HashSet<String> idSet = UserIdSet.getUserNameSet();
	
		@Override
		public boolean isValidRecord(int sourceIndex, String[] segment) {
			return idSet.contains(segment[5].toLowerCase());
		}
	}
	
	@Override 
	public Configuration getConf() {
		return null;
	}
	
	@Override
	public void setConf(Configuration conf) {
		
	}
	/*
	 * (non-Javadoc)
	 * ExtractUserByNameed Table
	 * UserFiltered Table
	 * Path: /group/tbsc-dev/xzwang/clickprediction/$date/ExtractUserByName/
	 *    1. Acookie的cookie id
	 *    2. 用户id (数字id)
	 *    3. 请求时间
	 *    4. 用户id（32位） 
	 *    5. query
	 *    6. 类目【直达后的类目】
	 *    7. 属性串
	 *    8. 搜索的排序方式，格式以转换
	 *    9. 展现的商品id【格式为豆腐块商品数字id list（用|分隔）+主搜索的商品数字id list（id之间用|分隔），如果只有豆腐块商品，则为”豆腐块商品数字id list+”，如果只有主搜索的商品，则为“主搜索商品id list” 
	 * */	 

	@Override
	public int run(String[] args) throws Exception {
		String salerFile = DumpSaler.getSalerPath(args[0]);
		String output = getExtractUserByNamePath(args[0]);
		System.out.println("start job ExtractUserByName...");
		System.out.println("salerFile: " + salerFile);
		System.out.println("output: " + output);
		System.out.println("Continue(0/1)?");
		Scanner scanner = new Scanner(System.in);
		int c = scanner.nextInt(); 
		if (c == 0)
			return -1;
		DumpTableRunner r = new DumpTableRunner();
		r.addSourceTable(salerFile, "UTF-8", "\t");
		StringBuilder column = new StringBuilder();
		for (int i = 1; i < 29; ++i)
			column.append("1." + i + ", ");
		column.append("1.29");
		r.setResultFields(column.toString());
		r.setSourceRecordFilterClass(SourceFilter.class);
		r.setResultTable(output);
		r.run("ExtractUserByNameJob");
		return 0;		 	
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		if (args.length != 1) {
			System.out.println("Usage: hadoop jar xx.jar <date>.");
			System.exit(-1);
		}
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new ExtractUserByName(), args);
		System.exit(res);
	}
}
